# coding=utf-8
import sys
from os import path
sys.path.append(path.dirname(path.dirname(path.abspath(__file__))))
try:
    import findspark
    findspark.init()
except ImportError as err:
    print("cannot import findspark, skipped...")
import pyspark


print("file abspath=%s" % path.abspath(__file__))
print(sys.path)
from pyspark.sql.functions import regexp_replace, explode, udf, mean, max, min, input_file_name, concat_ws, split, lit, broadcast
from pyspark.sql.types import DecimalType, IntegerType, BooleanType, DoubleType

### Setup
from common.common_emr import CommonEMR

commonEmr = CommonEMR("ProductSummary product extract on EMR")

if commonEmr.args.env:
    env = commonEmr.args.env

input_parameters = commonEmr.input_parameters
department = input_parameters["department"]
print("input department is %s" % department)

### Execution
sqlContext = pyspark.SQLContext(commonEmr.sc)

df = commonEmr.sc.parallelize([
    # ['B000XO2VHI', '$12.45', True, 'Creative Teaching Press Spanish Basic Skills, 5-Chart Pack (5792)'],
    # ['B01AXBK2GW', '$10.39', False, 'Teacher Created Resources Calling All Superheroes Mini Bulletin Board (5825)'],
    ['B000XO2VHI', '$12.45', True, 'Creative Teaching Press Spanish Basic Skills, 5-Chart Pack (5792)'],
    ['B003EWFICQ', '$5.60', True, 'Teacher Created Resources Cursive Mini Bulletin Board, Black (4099)']
    ]).\
    toDF(["asin","price","is_prime","name"])

df2 = commonEmr.sc.parallelize([
    # ['1936023334', '$4.99', False, 'Reading Comprehension, Grade 5 (Skill Builders)'],
    # ['159673292X', '$21.80', False, 'Critical & Creative Thinking Activities, Grade 1'],
    ['B000XO2VHI', '$12.45', True, 'Creative Teaching Press Spanish Basic Skills, 5-Chart Pack (5792)']
    ]).\
    toDF(["asin","price","is_prime","name"])

# df.select(df.asin.alias("origin_asin")).intersect(df2.select(df2.asin.alias("new_asin")))


if env == "testing":
    print("this is testing env.")
    df.show()
    df.printSchema()
    df2.show()
    df2.printSchema()
else:
    df.show()
    df.printSchema()
    df2.show()
    df2.printSchema()

### spark-submit --master local[4] --py-files s3://pt-emr-common/SourceCode/common_emr.zip ./testing_df_intersect_except.py --env testing > out

### result
# displaying df_union...
# +----------+------+--------+--------------------+
# |      asin| price|is_prime|                name|
# +----------+------+--------+--------------------+
# |B000XO2VHI|$12.45|    true|Creative Teaching...|
# |B003EWFICQ| $5.60|    true|Teacher Created R...|
# |B000XO2VHI|$12.45|    true|Creative Teaching...|
# +----------+------+--------+--------------------+
#
# displaying origin_asin...
# +-----------+
# |origin_asin|
# +-----------+
# | B000XO2VHI|
# | B003EWFICQ|
# +-----------+
#
# displaying new_asin...
# +----------+
# |  new_asin|
# +----------+
# |B000XO2VHI|
# +----------+
#
# displaying intersect_asin...
# +-----------+
# |origin_asin|
# +-----------+
# | B000XO2VHI|
# +-----------+
#
# displaying except_asin...
# +--------+
# |new_asin|
# +--------+
# +--------+
#
# +----+-----+--------+----+--------+
# |asin|price|is_prime|name|new_asin|
# +----+-----+--------+----+--------+
# +----+-----+--------+----+--------+